package cn.dmp.tags

import cn.dmp.utils.{JedisPools, TagsUtils}
import com.typesafe.config.ConfigFactory
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.{HColumnDescriptor, HTableDescriptor, TableName}
import org.apache.hadoop.hbase.client.{Admin, Connection, ConnectionFactory, Put}
import org.apache.hadoop.hbase.io.ImmutableBytesWritable
import org.apache.hadoop.hbase.mapred.TableOutputFormat
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.mapred.JobConf
import org.apache.spark.SparkContext
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataset, SparkSession}

import scala.collection.mutable.ListBuffer

/**
  * 需求：数据标签化变成: (用户id，用户的标签化集合)
  * 将结果输出到文件 和 hbase数据库中
  *
  */
object Tags4Ctx {
  def main(args: Array[String]): Unit = {


    if (args != 5){
      println(
        """
          |cn.dmp.tags.Tags4Ctx
          |参数：
          |   logInputPath
          |   dictFilePath
          |   停用词库
          |   日期
          |   resultOutputPath
        """.stripMargin)
    }

    val Array(logInputPath, dictFilePath, stopWordsFilePath, day, outputPath) = args

    val sparkSession: SparkSession = SparkSession.builder().appName("Tags4Ctx").master("local[*]")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer") //spark优化：spark.serializer修改磁盘序列化方式
      .enableHiveSupport()
      .getOrCreate()
    val sc: SparkContext = sparkSession.sparkContext

    // 字典文件---appMapping
    val dictMap: Map[String, String] = sparkSession.read.textFile(dictFilePath).map(line => {
      val fields: Array[String] = line.split("\t", -1)
      (fields(4), fields(1))
    }).collect().toMap

    // 停止词库字典
    val stopWordsMap: Map[String, Int] = sparkSession.read.textFile(stopWordsFilePath).map((_, 0)).collect().toMap

    // 将字典数据广播executor，不然mao里获取不到
    val broadcastAppDict: Broadcast[Map[String, String]] = sc.broadcast(dictMap)
    val broadcastStopWordsDict: Broadcast[Map[String, Int]] = sc.broadcast(stopWordsMap)

    val load = ConfigFactory.load()//这个ConfigFactory类是typesafe.config包下的。其.load()方法会读取项目resources目录下的application.conf文件，若不存在会读其他的..文件.。
    val hbTableName: String = load.getString("hbase.table.name")

    // 因为最终有要求要存储到hbase中，因此要先创建hbase表：判断hbase表是否存在，不存在则创建。
    // 先要连接到hbase的zookeeper地址。ConnectionFactory.createConnection()创建一个连接，连接里面是zk的地址
    val configuration: Configuration = sc.hadoopConfiguration
    configuration.set("hbase.zookeeper.quorum", load.getString("hbase.zookeeper.host"))
    val hbConn: Connection = ConnectionFactory.createConnection(configuration)
    val hbAdmin: Admin = hbConn.getAdmin

    if (!hbAdmin.tableExists(TableName.valueOf(hbTableName))){
      println(s"$hbTableName 不存在...")
      println(s"正在创建 $hbTableName 表...")
      // 建hbase表需要先创建 表名描述信息（填表名）,还有列簇Colunmn描述信息（填列簇名）
      val hTableDescriptor = new HTableDescriptor(TableName.valueOf(hbTableName))
      // 这里面填写列簇名。这里的列簇名我给设置为动态传递参数来的，比如日期。
      val columnDescriptor = new HColumnDescriptor(s"day$day")
      // 表描述对象 添加 列簇描述对象。
      hTableDescriptor.addFamily(columnDescriptor)
      // 开始创建。创建必须要先获得Admin才可以用其.createTable方法(表描述对象)
      hbAdmin.createTable(hTableDescriptor)

      //释放连接
      hbAdmin.close()
      hbConn.close()
    }

    // 得到JobConf才能去写到hbase中。指定key的输出类型
    val jobConf = new JobConf(configuration)
    // jobConf要设置一些参数：输出是什么类型：表；数据存入哪张表即指定表名称
    jobConf.setOutputFormat(classOf[TableOutputFormat])
    jobConf.set(TableOutputFormat.OUTPUT_TABLE, hbTableName)


    // 读取日志parquet文件，这15个id只要有一个不为空
    val ds: Dataset[(String, List[(String, Int)])] = sparkSession.read.parquet(logInputPath)
      .where(TagsUtils.hasSomeUserIdCondition)
      .mapPartitions( par => {
        val jedis = JedisPools.getJedis()
        val result: Iterator[(String, List[(String, Int)])] = par.map(row => {
          // 行数据进行标签化处理
          // 1.调用Ads打标
          val ads: Map[String, Int] = Tags4_Ads.makeTags(row)
          // 2.调用App打标
          val apps: Map[String, Int] = Tags4_App.makeTags(row, broadcastAppDict.value)
          // 3.调用Devices打标
          val devices: Map[String, Int] = Tags4_Devices.makeTags(row)
          // 4.调用KeyWords打标
          val keywords: Map[String, Int] = Tags4_KeyWords.makeTags(row, broadcastStopWordsDict.value)

          // 5. 商圈的标签
          val busines = Tags4Business.makeTags(row, jedis)


          // 得到所有用户id
          val allUserId: ListBuffer[String] = TagsUtils.getAllUserId(row)

          (allUserId(0), (ads ++ apps ++ devices ++ keywords ++ busines).toList) //Map数据之间相加是 ++)
        })

        jedis.close()
        result
      })

    val res: RDD[(String, List[(String, Int)])] = ds.rdd.reduceByKey((a, b) => {
      //同一个人(a ++ b)后为：List(("K电视剧",1),("APP爱奇艺",1) ("K电视剧",1))，所以还需要在按_._1来分groupBy变成Map["K电视剧",List(("K电视剧",1),("K电视剧",1))]，然后在加
      //写法一：(a ++ b).groupBy(_._1).mapValues(_.foldLeft(0)(_ + _._2)).toList //或者.mapValues(_.length).toList
      //写法二：map偏函数
      //      (a ++ b).groupBy(_._1).map{
      //        case (k, sameTags) => (k, sameTags.map(_._2).sum)
      //      }.toList
      (a ++ b).groupBy(_._1).mapValues(_.length).toList
    })
    // 最终写到文件里,eg其中一条数据如下:
    // (AD:1230234, List((河南省,2), (洛阳市,2), (K电视剧，4),(APP爱奇艺,1), (地域方面的kv), (广告方面的kv), ... ))
    res.saveAsTextFile(outputPath)

    /*
    // 将数据写到hbase中，那么res数据必须按照hbase格式化一下。
    res.map{//这个map算子后直接跟{}而没有()就是偏函数，偏函数里用case！。
      case (userId, userTags) => {
        //这个Put类是都要导hbase包。Bytes也要是Hbase的包。Put的参数是rowKey,是字节
        val put = new Put(Bytes.toBytes(userId))

        // 将每一个元祖变成一个字符串，mkString(",")将条数据以","隔开，最终合成一个字符串,
        val tags: String = userTags.map(t => t._1+":"+t._2).mkString(",")

        // 将tags与put关联。指定列簇:Array[Byte]和列限定符:Array[Byte]，和值values:Array[Byte]
        // 所以要先创建一个hbase表！！！！！在前面创建过了。然后对该表添加列簇
        put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes(s"day$day"), Bytes.toBytes(tags))

        // 想把数据存到hbase中，那么数据个格式必须是对偶元祖
        (new ImmutableBytesWritable(), put)   //ImmutableBytesWritable修饰的是rowKey对象
      }
    }.saveAsHadoopDataset(jobConf)  //真正存储到hbase数据库中
    */
    res.saveAsTextFile(outputPath)


    sparkSession.stop()
    sc.stop()

    //运行：五个参数



  }
}
